package com.sunzm.flink.datastream.scala.source.kafka.deserializer

import java.util.Properties

import com.alibaba.fastjson.JSONObject
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.connector.kafka.source.{KafkaSource, KafkaSourceOptions}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.kafka.clients.consumer.{ConsumerConfig, OffsetResetStrategy}
import org.apache.kafka.common.serialization.StringDeserializer

/**
 *
 * ${DESCRIPTION}
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-12-24 0:11
 */
object MyKafkaRecordDeserializationSchemaJob {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val properties: Properties = new Properties
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.200:9092")
    properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g-flink-test")
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")

    properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)
    properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
    properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
    properties.setProperty(KafkaSourceOptions.COMMIT_OFFSETS_ON_CHECKPOINT.key(), "true")

    val kafkaSource = KafkaSource.builder[JSONObject]()
      .setProperties(properties)
      .setBootstrapServers(properties.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))
      .setTopics("test")
      .setGroupId(properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG))
      .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
      .setDeserializer(new MyKafkaRecordDeserializationSchema)
      .build()

    val dataDS = env.fromSource(kafkaSource,
      WatermarkStrategy.noWatermarks(),
      "KafkaSource")
      .uid("kafka-data-source-20211224001")

    dataDS.print("data")

    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
